package com.test.finkcdc.job;

import com.test.finkcdc.config.SourceConfig;
import com.test.finkcdc.entity.JobInfo;
import com.test.finkcdc.entity.JsonRecord;
import com.test.finkcdc.util.*;
import com.ververica.cdc.connectors.base.options.StartupOptions;
import com.ververica.cdc.connectors.oracle.OracleSource;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase;
import org.apache.flink.util.OutputTag;

import java.util.*;
import java.util.stream.Collectors;

public class CommonSource {
    public static <T> void init(List<JobInfo> jobInfos) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.getCheckpointConfig().setCheckpointInterval(1800000L);
        env.getCheckpointConfig().setCheckpointTimeout(300000L);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(5)));
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        Properties pro = new Properties();
        pro.setProperty("decimal.handling.mode", "STRING");
        pro.setProperty("log.mining.strategy", "online_catalog");
        pro.setProperty("log.mining.continuous.mine", "true");
        pro.setProperty("log.mining.batch.size.min", "20000");
        pro.setProperty("log.mining.batch.size.max", "3000000");
        pro.setProperty("log.mining.sleep.time.increment.ms", "0");
        pro.setProperty("log.mining.sleep.time.default.ms", "0");
        pro.setProperty("log.mining.sleep.time.max.ms", "0");
        pro.setProperty("log.mining.dml.parser", "legacy");
        pro.setProperty("database.serverTimezone", "Asia/Shanghai");
        pro.setProperty("database.tablename.case.insensitive", "false");
        pro.setProperty("scan.startup.mode", "inital");
        List<String> tableNames = new ArrayList<>();
        jobInfos.stream().forEach(info -> {
            tableNames.add(info.getTableName().toUpperCase());
        });
        String tableList = String.join(",", tableNames.stream().map(c -> SourceConfig.USER_NAME + "." + c).collect(Collectors.toList()));
        SourceFunction<JsonRecord> sourceFunction = OracleSource.<JsonRecord>builder()
                .hostname(SourceConfig.HOST_NAME)
                .url(SourceConfig.URL)
                .database(SourceConfig.DATABASE)
                .tableList(tableList)
                .username(SourceConfig.USER_NAME)
                .password(SourceConfig.PASSWORD)
                .deserializer(new JsonStringDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.latest())
                .debeziumProperties(pro)
                .build();
        DataStreamSource<JsonRecord> input = env.addSource(sourceFunction);
        Map<String, OutputTag<JsonRecord>> outputTagMap = new HashMap<>();
        for (String tableName : tableNames) {
            outputTagMap.put(tableName, new OutputTag<JsonRecord>(tableName, TypeInformation.of(JsonRecord.class)));
        }

        SingleOutputStreamOperator<Object> process = input.process(new OutputTableProcessFunc(outputTagMap)).name("名字");
        OutputTag<JsonRecord> create = new OutputTag<JsonRecord>("create", TypeInformation.of(JsonRecord.class));
        OutputTag<JsonRecord> update = new OutputTag<JsonRecord>("update", TypeInformation.of(JsonRecord.class));
        OutputTag<JsonRecord> delete = new OutputTag<JsonRecord>("delete", TypeInformation.of(JsonRecord.class));

        jobInfos.stream().forEach(info -> {
            DataStream<JsonRecord> dataStream = process.getSideOutput(outputTagMap.get(info.getTableName()));
            SingleOutputStreamOperator<Object> sub=dataStream.process(new OutputTableSubProcessFun(create,update,delete)).setParallelism(3);
            if(StringUtils.isNoneEmpty(info.getInsertSql())){
                sub.getSideOutput(create).addSink(new InsertSink(info).getSinkFunction()).setParallelism(2).name(info.getTableName()+"_insert");
            }
            if(StringUtils.isNoneEmpty(info.getUpdateSql())){
                sub.getSideOutput(update).addSink(new UpdateSink(info)).setParallelism(2).name(info.getTableName()+"_update");

            }
            if(StringUtils.isNoneEmpty(info.getDeleteSql())){
                sub.getSideOutput(update).addSink(new DeleteSink(info)).setParallelism(2).name(info.getTableName()+"_delete");
            }

        });
        env.execute("job_name");
    }
}
